package ds_recommended

import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.{ALS, ALSModel}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, explode, to_json}
import org.apache.spark.sql.types.{ArrayType, FloatType, IntegerType, StructField, StructType}

import java.util.Properties

object user_to_recommended {
  def main(args: Array[String]): Unit = {

    //  准备spark的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("基于协同过滤的推荐")
      .getOrCreate()


    //  导入隐式转换，避免dataframe转化dataset的数据类型的不匹配的报错
    import spark.implicits._

    //  连接mysql的配置
    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  加载订单数据并选择需要用到的字段
    //  由于转换数据类型的时候会出现空值，下面做一个过滤
    val data=spark.read
      .jdbc("jdbc:mysql://192.168.67.193:3307/dwd?useSSL=false","order_info",connect)
      .select("user_id","product_id","rating")
      .withColumn("user_id",col("user_id").cast("long"))
      .withColumn("product_id",col("product_id").cast("long"))
      .withColumn("rating",col("rating").cast("double"))
      .filter(col("user_id").isNotNull)
      .filter(col("product_id").isNotNull)
      .filter(col("rating").isNotNull)



    //  构建als协同过滤的隐式反馈    implicit:隐式的
    /*
        setRegParam(0.1)：这个参数设置了 正则化参数（regParam），它用于控制模型复杂度，防止过拟合。在矩阵分解过程
        中，regParam 用于惩罚模型中的大权重值，从而避免模型过度依赖某些特定的特征，导致过拟合。setRegParam(0.1) 表示
        正则化项的系数为 0.1，较小的值允许模型有更多的自由度，较大的值则加强了正则化，限制了模型的复杂度。
     */

    val alsimplicit=new ALS()
      .setMaxIter(10)
      .setRegParam(0.1)
      .setImplicitPrefs(true)
      .setUserCol("user_id")
      .setItemCol("product_id")
      .setRatingCol("rating")
      .setColdStartStrategy("drop")


    //  根据设置构建模型
    val model:ALSModel=alsimplicit.fit(data)


    //  给每个用户推荐前10个商品
    val user_recommend=model.recommendForAllUsers(10)
    //  查看每个用户推荐的10个商品
    user_recommend.limit(50).show(false)


    //  由于jdbc只支持基本的数据类型，不支持array<struct<product_id:int, rating:float>>:[{74, 0.59754586}, {215, 0.5829263}, {107, 0.529323},
    //  {108, 0.52268565}, {319, 0.5159624}, {81, 0.51302}, {232, 0.5121461}, {67, 0.5087154},
    //  {127, 0.5055256}, {257, 0.48255214}] 这样的复合数据类型，所以这一列的数据需要进行处理一下

    user_recommend.createOrReplaceTempView("result")

    val result_recommend=spark.sql(
      """
        |select
        |user_id,
        |replace(replace(cast(recommendations as string),"{",""),"}","") as recommendations
        |from result
        |""".stripMargin)

    //  将推荐的结果存放到mysql里面
    result_recommend
      .withColumn("user_id",col("user_id").cast("string"))
      .write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.67.193:3307/dws?useSSL=false","offline_recommend",connect)



    spark.close()
  }

}
